package com.lcc.realtime.util;

import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import javax.annotation.Nullable;
import java.nio.charset.StandardCharsets;
import java.util.Properties;

import static com.lcc.realtime.common.Constant.*;

public class FlinkSinkUtil {
    public static SinkFunction<String> getKafkaSink(String topic){
        Properties prop = new Properties();
        prop.put(BOOTSTRAP_SERVERS,KAFKA_BROKERS);
        prop.put("transaction.timeout.ms",15*60*1000);//事务超时时间

        return
                new FlinkKafkaProducer<String>(
                "default", //默认topic
                new KafkaSerializationSchema<String>() { //序列化器
                    @Override
                    public ProducerRecord<byte[], byte[]> serialize(String s,
                                                                    @Nullable Long aLong) {
                        return new ProducerRecord<>(topic,s.getBytes(StandardCharsets.UTF_8));
                    }
                },
                prop, // 配置
                FlinkKafkaProducer.Semantic.EXACTLY_ONCE //写入模式，精确一次
        )
        ;
    }
}
